package com.scalyze.apps.socialanyltics.collector;

import java.util.concurrent.LinkedBlockingQueue;

import com.scalyze.apps.socialanyltics.datastore.JSONService;


public class Dequeuer implements Runnable {

	private LinkedBlockingQueue<String> messageQueue = new LinkedBlockingQueue<String>();
	private String hbaseMasterNode, hbaseMasterPort,zookeeperHost, trendName;
        private JSONService service; 

	public Dequeuer(LinkedBlockingQueue<String> messageQueue,
			String hbaseMasterNode, String hbaseMasterPort,String zookeeperHost, String trendName, JSONService service) {
		this.messageQueue = messageQueue;
		this.hbaseMasterNode = hbaseMasterNode;
		this.hbaseMasterPort = hbaseMasterPort;
		this.zookeeperHost=zookeeperHost;
		this.trendName = trendName;
                this.service = service;
	}

	public void run() {
		while (!Thread.interrupted()) {
			try {
				String message = messageQueue.take();
				System.out.println(message);
				service.parseMessage(message, hbaseMasterNode, hbaseMasterPort,zookeeperHost, trendName);
				// // JSONObject messageJSON = new JSONObject(message);
				//
				// // ignore delete records for now
				// if (messageJSON.has("delete")) {
				// continue;
				// }
				//
				// // create a copy with some renamed fields
				// JSONObject statusJSON = getStatus(messageJSON);
				// Message m = new Message(streamName,
				// "io.s4.example.twittertopiccount.Status",
				// statusJSON.toString());
				// driver.send(m);
			} catch (InterruptedException ie) {
				Thread.currentThread().interrupt();
				// } catch (IOException ioe) {
				// ioe.printStackTrace();
				// } catch (JSONException je) {
				// je.printStackTrace();
				// Thread.currentThread().interrupt();
			}
		}
	}

	// public JSONObject getStatus(JSONObject origStatusJSON) {
	// try {
	// if (origStatusJSON == null || origStatusJSON.equals(JSONObject.NULL)) {
	// return null;
	// }
	//
	// JSONObject statusJSON = new JSONObject();
	// copyFields(origStatusJSON, statusJSON, fieldMap);
	//
	// JSONObject origUserJSON = origStatusJSON.getJSONObject("user");
	// JSONObject userJSON = new JSONObject();
	// copyFields(origUserJSON, userJSON, fieldMap);
	// statusJSON.put("user", userJSON);
	//
	// return statusJSON;
	// } catch (Exception e) {
	// Logger.getLogger("s4").error(e);
	// }
	//
	// return new JSONObject();
	// }
	//
	// public void copyFields(JSONObject from, JSONObject to, Map<String,String>
	// fieldMap)
	// throws JSONException {
	// for (Iterator<?> it = from.keys(); it.hasNext(); ) {
	// String fieldName = (String) it.next();
	// String adjustedFieldName = fieldName;
	// String proposedFieldName = null;
	// if ((proposedFieldName = fieldMap.get(fieldName)) != null) {
	// adjustedFieldName = proposedFieldName;
	// }
	// to.put(adjustedFieldName, from.get(fieldName));
	// }
	// }
}